# SPDX-License-Identifier: Apache-2.0

from datetime import datetime

import pretend
import pytest

from pyramid.httpexceptions import HTTPSeeOther

from warehouse.admin.views import malware_reports as views
from warehouse.packaging.models import LifecycleStatus, Project

from ....common.db.accounts import UserFactory
from ....common.db.packaging import (
    ProjectFactory,
    ProjectObservationFactory,
    RoleFactory,
)


class TestMalwareReportsList:
    def test_malware_reports_list(self, db_request):
        assert views.malware_reports_list(db_request) == {"malware_reports": []}

    def test_malware_reports_list_with_observations(self, db_request):
        ProjectObservationFactory.create(kind="is_spam")
        ProjectObservationFactory.create(kind="is_malware", actions={"foo": "bar"})
        ProjectObservationFactory.create(kind="is_malware", related=None)
        malware = ProjectObservationFactory.create_batch(size=3, kind="is_malware")

        malware_reports = views.malware_reports_list(db_request)["malware_reports"]
        assert {m.id for m in malware_reports} == {m.id for m in malware}


class TestMalwareReportsProjectList:
    def test_malware_reports_project_list(self, db_request):
        project = ProjectFactory.create()
        assert views.malware_reports_project_list(project, db_request) == {
            "project": project,
            "malware_reports": [],
        }

    def test_malware_reports_project_list_with_project(self, db_request):
        project = ProjectFactory.create()
        report = ProjectObservationFactory.create(kind="is_malware", related=project)

        assert views.malware_reports_project_list(project, db_request) == {
            "project": project,
            "malware_reports": [report],
        }

    def test_malware_reports_project_list_with_project_and_actions(self, db_request):
        project = ProjectFactory.create()
        ProjectObservationFactory.create(
            kind="is_malware", related=project, actions={"foo": "bar"}
        )

        assert views.malware_reports_project_list(project, db_request) == {
            "project": project,
            "malware_reports": [],
        }

    def test_malware_reports_project_verdict_not_malware(self, db_request):
        project = ProjectFactory.create()
        report = ProjectObservationFactory.create(kind="is_malware", related=project)
        ProjectObservationFactory.create(kind="something_else", related=project)

        db_request.POST["confirm_project_name"] = project.name
        db_request.POST["reason"] = "This is a test"
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.malware_reports_project_verdict_not_malware(project, db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(
                f"Project {project.name} marked as not malware.\n"
                "Please update related Help Scout conversations.",
                queue="success",
            )
        ]

        assert len(report.actions) == 1
        action_record = list(report.actions.values())[0]

        assert action_record["action"] == "verdict_not_malware"
        assert action_record["actor"] == db_request.user.username
        assert isinstance(datetime.fromisoformat(action_record["created_at"]), datetime)
        assert action_record["reason"] == "This is a test"

    def test_malware_reports_project_verdict_quarantine(self, db_request):
        owner_user = UserFactory.create(is_frozen=False)
        project = ProjectFactory.create()
        RoleFactory(user=owner_user, project=project, role_name="Owner")
        report = ProjectObservationFactory.create(kind="is_malware", related=project)

        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.malware_reports_project_verdict_quarantine(project, db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(
                f"Project {project.name} quarantined.\n"
                "Please update related Help Scout conversations.",
                queue="success",
            )
        ]

        assert project.lifecycle_status == LifecycleStatus.QuarantineEnter
        assert project.lifecycle_status_changed is not None
        assert (
            project.lifecycle_status_note
            == f"Quarantined by {db_request.user.username}."
        )
        assert len(report.actions) == 0
        assert owner_user.is_frozen is True

    def test_malware_reports_project_verdict_remove_malware(self, db_request):
        owner_user = UserFactory.create(is_frozen=False)
        project = ProjectFactory.create()
        RoleFactory(user=owner_user, project=project, role_name="Owner")
        report = ProjectObservationFactory.create(
            kind="is_malware",
            related=project,
            additional={
                "helpscout_conversation_url": "https://example.com/conversation/123"
            },
        )

        db_request.POST["confirm_project_name"] = project.name
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.malware_reports_project_verdict_remove_malware(
            project, db_request
        )

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(f"Deleted the project '{project.name}'", queue="success"),
            pretend.call(
                f"Malware Project {project.name} removed.\n"
                "Please update related Help Scout conversations.",
                queue="success",
            ),
        ]

        assert len(report.actions) == 1
        assert db_request.db.get(Project, project.id) is None
        assert owner_user.is_frozen is True
        assert owner_user.observations[0].kind == "account_abuse"


class TestMalwareReportsDetail:
    def test_malware_reports_detail(self, db_request):
        assert views.malware_reports_detail(db_request) == {"report": None}

    def test_malware_reports_detail_with_report(self, db_request):
        report = ProjectObservationFactory.create(kind="is_malware")
        db_request.matchdict["observation_id"] = str(report.id)

        assert views.malware_reports_detail(db_request) == {"report": report}

    def test_detail_not_malware_for_project(self, db_request):
        report = ProjectObservationFactory.create(kind="is_malware")
        db_request.matchdict["observation_id"] = str(report.id)
        db_request.POST["reason"] = "This is a test"
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.not_malware_for_project(db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(
                f"Project {report.related.name} marked as not malware.\n"
                "Please update related Help Scout conversations.",
                queue="success",
            )
        ]

        assert len(report.actions) == 1
        action_record = list(report.actions.values())[0]

        assert action_record["action"] == "verdict_not_malware"
        assert action_record["actor"] == db_request.user.username
        assert isinstance(datetime.fromisoformat(action_record["created_at"]), datetime)
        assert action_record["reason"] == "This is a test"

    def test_detail_verdict_quarantine_project(self, db_request):
        owner_user = UserFactory.create(is_frozen=False)
        project = ProjectFactory.create()
        RoleFactory(user=owner_user, project=project, role_name="Owner")
        report = ProjectObservationFactory.create(kind="is_malware", related=project)
        db_request.matchdict["observation_id"] = str(report.id)
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.verdict_quarantine_project(db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(
                f"Project {report.related.name} quarantined.\n"
                "Please update related Help Scout conversations.",
                queue="success",
            )
        ]

        assert report.related.lifecycle_status == LifecycleStatus.QuarantineEnter
        assert report.related.lifecycle_status_changed is not None
        assert report.related.lifecycle_status_note == (
            f"Quarantined by {db_request.user.username}."
        )
        assert len(report.actions) == 0
        assert owner_user.is_frozen is True

    def test_detail_remove_malware_for_project(self, db_request):
        owner_user = UserFactory.create(is_frozen=False)
        project = ProjectFactory.create()
        RoleFactory(user=owner_user, project=project, role_name="Owner")
        report = ProjectObservationFactory.create(
            kind="is_malware",
            related=project,
            additional={
                "helpscout_conversation_url": "https://example.com/conversation/123"
            },
        )

        db_request.matchdict["observation_id"] = str(report.id)
        db_request.POST["confirm_project_name"] = project.name
        db_request.POST["reason"] = "This is a test"
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.remove_malware_for_project(db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(f"Deleted the project '{project.name}'", queue="success"),
            pretend.call(
                f"Malware Project {report.related.name} removed and report updated",
                queue="success",
            ),
        ]

        assert len(report.actions) == 1
        assert db_request.db.get(Project, project.id) is None
        assert owner_user.is_frozen is True
        assert owner_user.observations[0].kind == "account_abuse"

    def test_detail_remove_malware_for_project_updates_all_observations(
        self, db_request
    ):
        owner_user = UserFactory.create(is_frozen=False)
        project = ProjectFactory.create()
        RoleFactory(user=owner_user, project=project, role_name="Owner")
        report = ProjectObservationFactory.create(
            kind="is_malware",
            related=project,
            additional={
                "helpscout_conversation_url": "https://example.com/conversation/123"
            },
        )
        # A second report for the same project
        report2 = ProjectObservationFactory.create(
            kind="is_malware",
            related=project,
            additional={
                "helpscout_conversation_url": "https://example.com/conversation/321"
            },
        )
        # A third report for the same project, but already has actions
        report3 = ProjectObservationFactory.create(
            kind="is_malware",
            related=project,
            actions={"foo": "bar"},
            additional={
                "helpscout_conversation_url": "https://example.com/conversation/987"
            },
        )
        # A Non-malware report
        report4 = ProjectObservationFactory.create(kind="is_spam", related=project)

        db_request.matchdict["observation_id"] = str(report.id)
        db_request.POST["confirm_project_name"] = project.name
        db_request.POST["reason"] = "This is a test"
        db_request.route_path = lambda a: "/admin/malware_reports/"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = UserFactory.create()

        result = views.remove_malware_for_project(db_request)

        assert len(report.actions) == 1
        assert len(report2.actions) == 1
        assert len(report3.actions) == 2
        assert len(report4.actions) == 0

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/admin/malware_reports/"
        assert db_request.session.flash.calls == [
            pretend.call(f"Deleted the project '{project.name}'", queue="success"),
            pretend.call(
                f"Malware Project {report.related.name} removed and report updated",
                queue="success",
            ),
        ]

    @pytest.mark.parametrize(
        "input_url",
        [
            "https://secure.helpscout.net/conversation/123456789/987654321",
            "https://secure.helpscout.net/conversation/123456789/23711?viewId=7857079#thread-9203145731"  # noqa: E501
            "https://api.helpscout.net/v2/conversations/123456789",
        ],
    )
    def test_add_helpscout_conversation_valid_url(self, db_request, input_url):
        report = ProjectObservationFactory.create(kind="is_malware")

        db_request.matchdict["observation_id"] = str(report.id)
        db_request.POST["helpscout_conversation_url"] = input_url
        db_request.route_path = lambda r, **kw: f"/admin/malware_reports/{report.id}"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        result = views.add_helpscout_conversation(db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == f"/admin/malware_reports/{report.id}"
        assert db_request.session.flash.calls == [
            pretend.call(
                f"HelpScout Conversation URL added to report for {report.related.name}",
                queue="success",
            )
        ]
        assert report.additional["helpscout_conversation_url"] == (
            "https://api.helpscout.net/v2/conversations/123456789"
        )

    def test_add_helpscout_conversation_invalid_url(self, db_request):
        report = ProjectObservationFactory.create(kind="is_malware")

        db_request.matchdict["observation_id"] = str(report.id)
        db_request.POST["helpscout_conversation_url"] = "invalid-url"
        db_request.route_path = lambda r, **kw: f"/admin/malware_reports/{report.id}"
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        result = views.add_helpscout_conversation(db_request)

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == f"/admin/malware_reports/{report.id}"
        assert db_request.session.flash.calls == [
            pretend.call(
                "Please provide a valid HelpScout Conversation URL.",
                queue="error",
            )
        ]
        assert "helpscout_conversation_url" not in report.additional
